package com.demo.realtime.job;

import com.demo.realtime.source.JbdcConnSingleton;
import com.demo.vo.PageLogsCount;
import com.demo.vo.PageLogsVo;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * @Author zengqingdong
 * @ClassName LanJuVistLogsJob
 * @Date 2020/8/8 1:51
 * @Version v1.0
 **/
public class LanJuVistLogsJob {

    private static final Logger logger = LoggerFactory.getLogger(LanJuVistLogsJob.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //StreamTableEnvironment tEnv = StreamTableEnvironment.create(streamEnv);
        streamEnv.enableCheckpointing(3000);
        CheckpointConfig checkpointConfig = streamEnv.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(6000);
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        RowTypeInfo rowTypeInfo = new RowTypeInfo(
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.DATE_TYPE_INFO);
        String sql = "select logId,domain,ip,url,userId,createTime from pagelogs  ";
        JdbcInputFormat.JdbcInputFormatBuilder jdbcInputFormatBuilder = JbdcConnSingleton.getInstance();
        JdbcInputFormat jdbcInputFormat = jdbcInputFormatBuilder.setQuery(sql).setRowTypeInfo(rowTypeInfo).finish();
        jdbcInputFormat.openInputFormat();
        DataStream dataStream = streamEnv.createInput(jdbcInputFormat);
        dataStream.map(new MapFunction() {
            @Override
            public Object map(Object o) throws Exception {
                return o.toString();
            }
        }).flatMap(new FlatMapFunction() {
            @Override
            public void flatMap(Object o, Collector collector) throws Exception {
                System.out.println("添加数据——"+o.toString());
                String arr[]  = o.toString().split(",");
                PageLogsCount pageLogsCount = new PageLogsCount(arr[4],arr[3],1);
                collector.collect(pageLogsCount);
            }
        }).returns(PageLogsCount.class).print();
        long startTime = System.currentTimeMillis();
        streamEnv.execute(LanJuVistLogsJob.class.getName());
        long endTime = System.currentTimeMillis();
        logger.info("程序运行时间：" + (endTime - startTime) / 1000 + "秒");
    }


}
